import { createReadStream, createWriteStream, existsSync } from 'fs';
import { createInterface } from 'readline';
import { dirname } from 'path';
import { mkdir } from 'fs/promises';
import { Worker, isMainThread, parentPort } from 'worker_threads';
import { scrapePoet } from './poet';

async function workerTask(url: string) {
  try {
    const result = await scrapePoet('https://aspoem.com' + url);
    return { success: true, data: result };
  } catch (error) {
    // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
    return { success: false, url, error };
  }
}

// 如果是Worker线程
if (!isMainThread) {
  // eslint-disable-next-line @typescript-eslint/no-misused-promises
  parentPort?.on('message', async (url) => {
    const result = await workerTask(url);
    parentPort?.postMessage(result);
  });
}

export async function batchScrapePoetUrls(
  inputFile: string,
  outputFile: string,
  delay: number = 1000,
  threadCount: number = 10,
) {
  // 检查输入文件是否存在
  if (!existsSync(inputFile)) {
    throw new Error(`Input file does not exist: ${inputFile}`);
  }

  // 确保输出目录存在
  const outputDir = dirname(outputFile);
  if (!existsSync(outputDir)) {
    await mkdir(outputDir, { recursive: true });
  }

  const inputStream = createReadStream(inputFile);
  const outputStream = createWriteStream(outputFile, { flags: 'a' });
  const rl = createInterface({ input: inputStream });

  let processedCount = 0;
  let errorCount = 0;
  const errorUrls: string[] = [];

  // 创建Worker池
  const workers = Array.from(
    { length: threadCount },
    () => new Worker(__filename, { workerData: { initialized: true } }),
  );

  // 处理Worker结果
  workers.forEach((worker) => {
    worker.on('message', ({ success, data, url, error }) => {
      if (success) {
        outputStream.write(JSON.stringify(data) + '\n');
        processedCount++;
      } else {
        errorUrls.push(url);
        errorCount++;
        console.error(`Error processing URL: ${url}`, error);
      }

      console.log(
        `Progress - Processed: ${processedCount}, Errors: ${errorCount}`,
      );
    });
  });
  // 添加Worker错误处理
  workers.forEach((worker) => {
    worker.on('error', (error) => {
      console.error('Worker error:', error);
    });
    worker.on('exit', (code) => {
      if (code !== 0) {
        console.error(`Worker stopped with exit code ${code}`);
      }
    });
  });

  // 分配任务给Worker
  let workerIndex = 0;
  for await (const url of rl) {
    workers[workerIndex % threadCount].postMessage(url);
    workerIndex++;

    // 添加随机延时
    const randomDelay = delay + (Math.random() * 2000 - 1000);
    await new Promise((resolve) => setTimeout(resolve, randomDelay));
  }

  // 关闭所有Worker
  // eslint-disable-next-line @typescript-eslint/no-misused-promises
  workers.forEach((worker) => worker.terminate());
  workers.length = 0;
  outputStream.end();

  console.log('Batch processing completed');
  if (errorUrls.length > 0) {
    console.log('Error URLs:', errorUrls);
  }
}
